package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/4/11
 *
        执行查询语句:  Table t = tableEnv.sqlQuery(sql)
                show
                select

        执行写操作语句:  tableEnvironment.executeSql(sql)
                insert
                create
 */
public class Demo2_HelloWorld_sql
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        Table table = tableEnvironment.fromDataStream(ds);

        //不起名写sql
        //Table result = tableEnvironment.sqlQuery("select * from " + table + " where id = 's1' ");

        //可以起名写sql
        tableEnvironment.createTemporaryView("t1",table);
        Table result = tableEnvironment.sqlQuery("select * from t1 where id = 's1' ");
        result.execute().print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }


    }
}
